[YAML] Add SplunkIO write SchemaTransform and WriteToSplunk by schizophrenicmaniac · Pull Request #40481 · apache/beam

schizophrenicmaniac · 2026-10-09T09:45:18Z

Adds a SchemaTransform for SplunkIO writes and exposes it in Beam YAML as WriteToSplunk. It follows the same approach as the Datadog one (#38362). SplunkIO in Java only supports writes, so there is no ReadFromSplunk here.

Each input row becomes a SplunkEvent. event (string) is required. time (int64), host, source, sourcetype, index and fields are optional. fields can be a nested row or a JSON object string, and other columns are ignored. The config covers everything SplunkIO.Write supports (url, token, batch_count, parallelism, disable_certificate_validation, root_ca_certificate_path, enable_batch_logs, enable_gzip_http_compression) plus error_handling. The error output has the same shape as Datadog's: rows that can't be converted come back with failed_row and a 400, and HEC write failures come back with the HTTP status and the payload. Without error_handling, a failed write fails the pipeline.

While testing this I found that SplunkWriteError loses all of its values once it goes through a coder. It has @DefaultSchema(AutoValueSchema.class), but its accessors aren't get-prefixed, so the inferred schema has no fields and every error decodes as null/null/null. SplunkEventWriterTest doesn't catch this because containsInAnyOrder decodes both sides with the same coder. I replaced the annotation with a hand-written SplunkWriteErrorSchemaProvider, so SplunkWriteError is still encoded as a row, now with statusCode, statusMessage and payload fields. This changes the output coder of SplunkIO.Write, so I listed it under bugfixes in CHANGES.md.

The YAML test runs SplunkIO on the Python FnApiRunner, which failed on the stateful writer's timers: the runner length prefixes coders it doesn't know for PCollections but not for timer family coders, so it tried to decode the Java VarIntCoder keys itself. The runner now length prefixes timer coders the same way. This also makes the beam:coders:javasdk:0.1 registration in integration_tests.py unnecessary, so that's removed.

Changes:

SplunkWriteSchemaTransformProvider / SplunkWriteSchemaTransformConfiguration, registered in the IO expansion service
WriteToSplunk in standard_io.yaml and the matching entry in standard_external_transforms.yaml
SplunkWriteErrorSchemaProvider, used as the default schema of SplunkWriteError and for the fields of the error output
FnApiRunner length prefixes timer family coders (translations.py)
Java tests against a mock HEC server, a YAML integration test with a fake HEC fixture (tests/splunk.yaml), and a translations test for the timer coders

Fixes #40470

codecov · 2026-10-09T10:06:48Z

Codecov Report

❌ Patch coverage is 3.70370% with 104 lines in your changes missing coverage. Please review.
✅ Project coverage is 58.83%. Comparing base (a8d6afd) to head (8b23369).
⚠️ Report is 75 commits behind head on master.

Files with missing lines	Patch %	Lines
...n/apache_beam/yaml/test_utils/splunk_test_utils.py	0.00%	103 Missing ⚠️
sdks/python/apache_beam/yaml/integration_tests.py	0.00%	1 Missing ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #40481       +/-   ##
=============================================
- Coverage     80.04%   58.83%   -21.22%     
- Complexity        0    13972    +13972     
=============================================
  Files           490     2601     +2111     
  Lines         82916   273357   +190441     
  Branches          0    11287    +11287     
=============================================
+ Hits          66373   160830    +94457     
- Misses        16543   106510    +89967     
- Partials          0     6017     +6017

Flag	Coverage Δ
python	`79.84% <3.70%> (-0.21%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

github-actions · 2026-10-09T10:43:21Z

Assigning reviewers:

R: @jrmccluskey for label python.
R: @kennknowles for label java.
R: @derrickaw for label yaml.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

derrickaw · 2026-10-12T19:59:52Z

waiting on author

…io-for

schizophrenicmaniac · 2026-10-13T08:01:39Z

Merged master and resolved the CHANGES.md conflict, keeping both entries.

derrickaw

A few comments on yaml related code.

… Splunk fixture

derrickaw · 2026-10-13T18:14:47Z

LGTM for yaml/python

@kennknowles, can you review the java part? Thanks

kennknowles · 2026-10-14T14:00:43Z

          .apply("Create KV pairs", CreateKeys.of(parallelism()))
-          .apply("Write Splunk events", ParDo.of(writer));
+          .apply("Write Splunk events", ParDo.of(writer))
+          .setCoder(SplunkWriteErrorCoder.of());


As you've probably noticed, the trouble with this is that it requires a special wrapper to be encoded as a row to be used by other languages. This is not such a terrible thing, but it would be better to do a hand-written schema coder so that SplunkWriteError is still encoded as a row, it just cannot use the @AutoValueSchema annotation to do it (though fixing the get-prefix limitation might be a one-shot prompt...)

This would look like making a new SchemaCoder(toFunction, fromFunction) where the to/from are just the same logic that you have in the SchemaTransform wrapper.

Does this work? Am I overlooking at issue?

Aside: if we were to switch to the new coder since the @AutoValueSchema annotation is not working as intended, we should replace it in the other file like so:

@DefaultCoder(SplunkWriteErrorCoder.class) public class SplunkWriteError { ... }

Then the coder will be automatic, here and everywhere.

But again, I think we can keep it encoded as a row so we don't have the breaking change and it works smoothly with the schema transform.

Yes, that works. I dropped SplunkWriteErrorCoder and added a SplunkWriteErrorSchemaProvider with hand-written to/from row functions (fields statusCode, statusMessage, payload). SplunkWriteError now uses it via @DefaultSchema instead of AutoValueSchema, so the setCoder call in SplunkIO.Write is gone and the output is inferred as a SchemaCoder like before, just with actual fields now. The schema transform still builds its own error rows since it adds failed_row.

I didn't touch the get-prefix handling in AutoValueSchema. That would change the inferred schema, and so the coder, of any other AutoValue type that has non-get accessors, so it seemed better kept out of this PR.

…io-for

…m coder

schizophrenicmaniac · 2026-10-16T09:19:41Z

@kennknowles I switched SplunkWriteError over to a hand-written schema provider as you suggested, could you take another look when you get a chance?

kennknowles

OK I took another look. Definitely an improvement! I see some other things that seem worth cleaning up.

kennknowles · 2026-10-21T13:31:59Z

+
+      Schema errorSchema =
+          Schema.builder()
+              .addNullableRowField("failed_row", inputRows.getSchema())


Is this duplicate with the SplunkWriteErrorSchemaProvider?

Partly. It's the same three fields plus failed_row. I changed it to take those fields from SplunkWriteErrorSchemaProvider.SCHEMA and build both kinds of error rows through the provider's toRow, so they're only defined in one place now.

kennknowles · 2026-10-21T13:32:42Z

+
+    assertEquals(
+        Schema.builder()
+            .addNullableField("statusCode", Schema.FieldType.INT32)


This seems like just a duplicate of the code in the schema provider. Is there something being tested here?

Not really. It only checks that the annotation picks up the provider, and the coder round trip and to/from row tests already cover that. Removed it.

kennknowles · 2026-10-21T13:34:56Z

+  # fallback URN. The payload is the snappy-compressed serialized Java coder,
+  # and the start of its class name stays uncompressed, so it can be used to
+  # tell the two apart.
+  if b'VarInt' in payload:


This seems like a major hack. What is the goal here? If there is a "standard coder" that is accidentally being serialized as beam:coders:javasdk then we should fix that in the Java SDK.

Agreed, it was a hack. The goal was to get the YAML test running on the Python FnApiRunner. SplunkIO keys its stateful writer with VarIntCoder, which has no portable URN, so it isn't a standard coder being serialized wrongly. The actual problem was in the runner: FnApiRunner length prefixes coders it doesn't know for PCollections, but not for timer family coders, so it tried to decode the Java timer keys itself. That's also why the BigEndianIntegerCoder registration for Datadog was there.

I changed setup_timer_mapping in translations.py to length prefix the timer coders the same way, so the runner treats those keys as bytes, and added a test for it. With that, the Splunk and Datadog YAML tests both pass without any beam:coders:javasdk:0.1 registration, so I removed it from integration_tests.py, the BigEndianIntegerCoder one included.

…rs in FnApiRunner

…io-for

schizophrenicmaniac · 2026-10-23T06:21:50Z

@kennknowles I've addressed your latest comments, could you take another look? The two red checks aren't from this change: BigQueryNestedRecordsIT hit a 503 backendError from BigQuery, and codecov/patch is flagging splunk_test_utils.py, which only runs in the YAML integration tests.

…io-for # Conflicts: # CHANGES.md

kennknowles

This looks pretty good. There are merge conflicts to be resolved, so that'll kick the tests off again anyhow. We'll get them green before merge.

[YAML] Add SplunkIO write SchemaTransform and WriteToSplunk

65fd4ed

github-actions Bot added python java io yaml labels Oct 9, 2026

github-actions Bot requested review from derrickaw, jrmccluskey and kennknowles October 9, 2026 10:43

github-actions Bot added the Next Action: Reviewers label Oct 9, 2026

[YAML] Decode Java VarIntCoder keys in integration test timers

3e1715b

github-actions Bot added Next Action: Author and removed Next Action: Reviewers labels Oct 12, 2026

Merge branch 'master' into fix/40470-feature-request-normalize-splunk…

22694d6

…io-for

github-actions Bot added Next Action: Reviewers and removed Next Action: Author labels Oct 13, 2026

derrickaw reviewed Oct 13, 2026

View reviewed changes

Comment thread sdks/python/apache_beam/yaml/test_utils/splunk_test_utils.py Outdated

Comment thread sdks/python/apache_beam/yaml/test_utils/splunk_test_utils.py

Comment thread sdks/python/apache_beam/yaml/test_utils/splunk_test_utils.py Outdated

[YAML] Return 400 for malformed payloads and verify after shutdown in…

ce71e14

… Splunk fixture

kennknowles requested changes Oct 14, 2026

View reviewed changes

schizophrenicmaniac added 2 commits October 15, 2026 14:53

Merge branch 'master' into fix/40470-feature-request-normalize-splunk…

9888038

…io-for

[Java] Give SplunkWriteError a hand-written schema instead of a custo…

ef96d29

…m coder

jrmccluskey removed their request for review October 19, 2026 15:07

derrickaw requested a review from kennknowles October 20, 2026 20:44

kennknowles requested changes Oct 21, 2026

View reviewed changes

[YAML] Reuse the SplunkWriteError schema and length prefix timer code…

d0027ed

…rs in FnApiRunner

github-actions Bot added the runners label Oct 21, 2026

Merge branch 'master' into fix/40470-feature-request-normalize-splunk…

5affd5c

…io-for

Merge branch 'master' into fix/40470-feature-request-normalize-splunk…

8b23369

…io-for # Conflicts: # CHANGES.md

kennknowles approved these changes Oct 23, 2026

View reviewed changes

Conversation

schizophrenicmaniac commented Oct 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

codecov Bot commented Oct 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

github-actions Bot commented Oct 9, 2026

Uh oh!

derrickaw commented Oct 12, 2026

Uh oh!

schizophrenicmaniac commented Oct 13, 2026

Uh oh!

derrickaw left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

derrickaw commented Oct 13, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

schizophrenicmaniac commented Oct 16, 2026

Uh oh!

kennknowles left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

schizophrenicmaniac commented Oct 23, 2026

Uh oh!

kennknowles left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

schizophrenicmaniac commented Oct 9, 2026 •

edited

Loading

codecov Bot commented Oct 9, 2026 •

edited

Loading